//
// Created by dave on 18-8-5.
//

#include <vector>
#include <glog/logging.h>
#include <evpp/event_loop.h>
#include <evpp/event_loop_thread_pool.h>
#include <evpp/tcp_server.h>
#include "jtt808_server.h"
#include "codec/data_pack.h"
#include "codec/data_utils.h"
#include "db/location.h"
#include "codec/platform/plat_common_resp.h"
#include "codec/terminal/location_report.h"
#include "codec/terminal/term_register.h"
#include "codec/platform/term_register_resp.h"

size_t _find_mark(const char* ptr, size_t start, size_t stop) {
    for (size_t i = start; i < stop; i++) {
        if (ptr[i] == 0x7E) {
            return i;
        }
    }
    return 0;
}

Jtt808Server::Jtt808Server(const std::string &listen_addr,const std::string &name, uint32_t acceptThreadNum, uint32_t childThreadNum) {
    acceptLoop_.reset(new evpp::EventLoop());
    tcpServer_ = std::make_unique<evpp::TCPServer>(acceptLoop_.get(), listen_addr, name, acceptThreadNum);
    tcpServer_->SetConnectionCallback(std::bind(&Jtt808Server::onConnection, this, std::placeholders::_1));
    tcpServer_->SetMessageCallback(std::bind(&Jtt808Server::onMessage, this, std::placeholders::_1, std::placeholders::_2));

//    childLoop_.reset(new evpp::EventLoop());
//    childPool_.reset(new evpp::EventLoopThreadPool(childLoop_.get(), childThreadNum));
//    childPool_->Start(true);
}

void Jtt808Server::start() {
    tcpServer_->Init();
    tcpServer_->Start();
    acceptLoop_->Run();
//    childLoop_->Run();
}

void Jtt808Server::onConnection(const evpp::TCPConnPtr &conn) {
    if (conn->IsConnected()) {
        conn->SetTCPNoDelay(true);
        conn->set_context(evpp::Any(std::make_shared<std::atomic_int>(0)));
    }
}

void Jtt808Server::onMessage(const evpp::TCPConnPtr &conn, evpp::Buffer *msg) {
    size_t size = msg->length();
    if (size > 4 * 1024) {
        msg->Reset();
        return;
    }

    const char* p = msg->data();
    size_t start = _find_mark(p, 0, size);
    if (start == 0 && p[0] != 0x7E) {
        msg->Reset();
    } else {
        do {
            size_t end = _find_mark(p, start + 1, size);
            if (end > 0) {
                if (end - start == 1) {
                    start = end;
                    continue;
                }

                onFrame(conn, p + start, end - start + 1);
//                childPool_->GetNextLoopWithHash(reinterpret_cast<uint64_t>(conn.get()))->QueueInLoop([this, conn, msg = std::string(&p[start], end - start + 1)] () mutable -> void {
//                    onFrame(conn, std::move(msg));
//                });
                start = _find_mark(p, end + 1, size);
                if (start == 0) {
                    msg->Skip(end + 1);
                    break;
                }
            } else {
                msg->Skip(start);
                break;
            }
        } while (true);
    }
}

void Jtt808Server::onFrame(const evpp::TCPConnPtr &conn, const char* ptr, size_t length) {
    int count = ++msgCounter_;
    if (count % 10000 == 0) {
        LOG(INFO) << "接收(万包)：" << count / 10000 << ", " << (clock() - startClock_) / (CLOCKS_PER_SEC / 1000);
    }

    if (length < 14) {
        LOG(WARNING) << "无效桢：" << length << "字节";
        return;
    }

    conn->Send(ptr, length);
//    DataPack dp = decode_data_pack(ptr, 0, length);
//    const auto &atomic = conn->context().Get<const std::shared_ptr<std::atomic_int>>();
//    auto seq = static_cast<unsigned short>(++*atomic);
//    if (seq >= 0xFFFF) {
//        atomic->store(0);
//        seq = 1;
//    }
//    if (dp.getMsgId() == 0x0200) {
//        auto lr = std::dynamic_pointer_cast<LocationReport>(dp.getBody());
//        // LOG(INFO) << dp.getSeqNo() << "|" << dp.getSim() << ": " << lr->getLng() << ", " << lr->getLat();
//        location loc(dp);
////        if (!saver_.save(loc)) {
////            LOG(WARNING) << "丢弃1条";
////        }
//
//        DataPack re;
//        re.setMsgId(0x8001);
//        re.setSeqNo(seq);
//        re.setSim(dp.getSim());
//
//        auto body = std::make_shared<PlatCommonResp>();
//        body->setTermMsgId(dp.getMsgId());
//        body->setTermSeqNo(dp.getSeqNo());
//        body->setResult(0);
//        re.setBody(body);
//
//        std::vector<char> vec = re.encode();
//        conn->Send(vec.data(), vec.size());
//    } else if (dp.getMsgId() == 0x0100) {
//        std::shared_ptr<TermRegister> tr = std::dynamic_pointer_cast<TermRegister>(dp.getBody());
//        // LOG(INFO) << dp.getSeqNo() << "|" << dp.getSim() << ": " << tr->getLicense();
//
//        DataPack re;
//        re.setMsgId(0x8100);
//        re.setSeqNo(seq);
//        re.setSim(dp.getSim());
//
//        auto resp = std::make_shared<TermRegisterResp>();
//        resp->setTermSeqNo(dp.getSeqNo());
//        resp->setResult(0);
//        resp->setAuth("abcdef");
//        re.setBody(resp);
//
//        std::vector<char> vec = re.encode();
//        conn->Send(vec.data(), vec.size());
//    } else {
//        LOG(INFO) << "ERROR: " << dp.getMsgId();
//    }
}
